package com.huan.rabbitmq.multi.config;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author huan.fu 2020/8/2 - 17:38
 */
@Configuration
@RequiredArgsConstructor
@Slf4j
public class MultiRabbitMqConfig {

    private final DefaultListableBeanFactory defaultListableBeanFactory;
    private final AutowireCapableBeanFactory autowireCapableBeanFactory;

    private static Map<String, RabbitProperties> multiMqPropertiesMap = new HashMap<String, RabbitProperties>() {
        {
            put("first", RabbitProperties.builder()
                    .host("47.101.130.164")
                    .port(5672)
                    .username("rabbit-multi-01")
                    .password("rabbit-multi-01")
                    .virtualHost("/rabbit-multi-01")
                    .queueName("queue-rabbit-multi-01").build());
            put("second", RabbitProperties.builder()
                    .host("47.101.130.164")
                    .port(5672)
                    .username("rabbit-multi-02")
                    .password("rabbit-multi-02")
                    .virtualHost("/rabbit-multi-02")
                    .queueName("queue-rabbit-multi-02").build());
        }
    };

    @PostConstruct
    public void initRabbitmq() {
        multiMqPropertiesMap.forEach((key, rabbitProperties) -> {

            AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(CachingConnectionFactory.class)
                    .addPropertyValue("cacheMode", CachingConnectionFactory.CacheMode.CHANNEL)
                    .addPropertyValue("host", rabbitProperties.getHost())
                    .addPropertyValue("port", rabbitProperties.getPort())
                    .addPropertyValue("username", rabbitProperties.getUsername())
                    .addPropertyValue("password", rabbitProperties.getPassword())
                    .addPropertyValue("virtualHost", rabbitProperties.getVirtualHost())
                    .getBeanDefinition();
            String connectionFactoryName = String.format("%s%s", key, "ConnectionFactory");
            defaultListableBeanFactory.registerBeanDefinition(connectionFactoryName, beanDefinition);
            CachingConnectionFactory connectionFactory = defaultListableBeanFactory.getBean(connectionFactoryName, CachingConnectionFactory.class);

            String rabbitAdminName = String.format("%s%s", key, "RabbitAdmin");
            AbstractBeanDefinition rabbitAdminBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(RabbitAdmin.class)
                    .addConstructorArgValue(connectionFactory)
                    .addPropertyValue("autoStartup", true)
                    .getBeanDefinition();
            defaultListableBeanFactory.registerBeanDefinition(rabbitAdminName, rabbitAdminBeanDefinition);
            RabbitAdmin rabbitAdmin = defaultListableBeanFactory.getBean(rabbitAdminName, RabbitAdmin.class);
            log.info("rabbitAdmin:[{}]", rabbitAdmin);

            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            defaultListableBeanFactory.registerSingleton(String.format("%s%s", key, "RabbitTemplate"), rabbitTemplate);

            SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
            // 设置监听的队列
            simpleMessageListenerContainer.setQueueNames(rabbitProperties.getQueueName());
            // 指定要创建的并发使用者的数量,默认值是1,当并发高时可以增加这个的数值，同时下方max的数值也要增加
            simpleMessageListenerContainer.setConcurrentConsumers(3);
            // 最大的并发消费者
            simpleMessageListenerContainer.setMaxConcurrentConsumers(10);
            // 设置是否重回队列
            simpleMessageListenerContainer.setDefaultRequeueRejected(false);
            // 设置签收模式
            simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            // 设置非独占模式
            simpleMessageListenerContainer.setExclusive(false);
            // 设置consumer未被 ack 的消息个数
            simpleMessageListenerContainer.setPrefetchCount(1);
            // 设置消息监听器
            simpleMessageListenerContainer.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
                try {
                    log.info("============> Thread:[{}] 接收到消息:[{}] ", Thread.currentThread().getName(), new String(message.getBody()));
                    log.info("====>connection:[{}]", channel.getConnection());
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    // 发生异常此处需要捕获到
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                }
            });
            /**
             * 1、simpleMessageListenerContainer.start();
             * 2、simpleMessageListenerContainer.stop();
             * 3、如果后期rabbitmq 的配置是从数据库中读取，即用户在页面上添加一个配置，就动态创建这个
             * 此时就需要调用 SimpleMessageListenerContainer#start 方法
             */
            String containerName = String.format("%s%s", key, "SimpleMessageListenerContainer");
            defaultListableBeanFactory.registerSingleton(containerName, autowireCapableBeanFactory.initializeBean(simpleMessageListenerContainer, containerName));
        });
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            RabbitTemplate firstRabbitTemplate = (RabbitTemplate) defaultListableBeanFactory.getBean("firstRabbitTemplate");
            firstRabbitTemplate.convertAndSend("exchange-rabbit-multi-01", "", "first queue message");
            log.info("first over...");
        }).start();
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            RabbitTemplate secondRabbitTemplate = (RabbitTemplate) defaultListableBeanFactory.getBean("secondRabbitTemplate");
            secondRabbitTemplate.convertAndSend("exchange-rabbit-multi-02", "", "second queue message");
            log.info("second over...");
        }).start();
    }
}
